package com.fourth.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.fourth.common.GmallConfig;
import com.fourth.utils.DimUtil;
import com.fourth.utils.DruidDSUtil;
import com.fourth.utils.JedisPoolUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Set;

/**
 * @Author CZQ
 * @Date 2022/7/19 16:06
 * @Version 1.0
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    private DruidDataSource druidDataSource;
    //构建连接池
    private JedisPool jedisPool;

    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource = DruidDSUtil.createDataSource();
        //初始化连接池
        jedisPool = JedisPoolUtil.getJedisPool();
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //获取连接
        DruidPooledConnection connection = druidDataSource.getConnection();


        //拼接SQL语句 upsert into db.tn(id,name,sex) values('1001','zs','male')
        String sinkTable = value.getString("sinkTable");
        JSONObject data = value.getJSONObject("data");
        String sql = genUpsertSql(sinkTable
                 , data);
        System.out.println(sql);

        //如果为更新操作，则先删除Redis中的数据
        /*if ("update".equals(value.getString("type"))){
            //获取连接池的连接
            Jedis jedis = jedisPool.getResource();
            DimUtil.delDimInfo(jedis,
                    sinkTable.toUpperCase(), //因为在Phoenix中表名都是大写
                    data.getString("id"));

            //连接归还下  连接池不用关在Open方法中构建一个并行度一个，但是连接需要归还到连接池中
            jedis.close();
        }*/

        //编译&执行  这里编译没有捕获异常是因为抛异常了，之所以声明：这里写如果数据写入失败少了一条数据，维表会关联不上任务会出错  如果不走MYSQL  所以应该让任务挂掉
        // 若不想挂掉则可以捕获异常，然后去访问MYSQL，
        // 还有一种思路就是如果访问Phoenix关联不上维表的时候，可以去访问一下MYSQL的话，

        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        preparedStatement.execute();
        connection.commit(); // 提交，因为Phoenix的连接不是自动提交的需要手动提交 建表DML默认不提交 前边是DDL默认提交

        //释放资源
        preparedStatement.close();
        connection.close();

    }

    // upsert into db.tn(id,name,sex) values('1001','zs','male')
    private String genUpsertSql(String sinkTable, JSONObject data) {
        //把(id,name,sex)当做key 把 ('1001','zs','male') 当做Value
        //获取key
        Set<String> columns = data.keySet();
        //获取Values
        Collection<Object> values = data.values();


        // StringUtils.join(columns, ",") => 相当于Scala中的columns.mkString(",")  => ["a","b","c"] -> "a,b,c"
        return "upsert into "+ GmallConfig.HBASE_SCHEMA+"." + sinkTable + "("+
                StringUtils.join(columns, ",") + ") values('"+StringUtils.join(values,"','")+"')" ;//这里对value中数据的','作为分隔符
    }
}
